package org.llc.flink.explore.table.wordcount

import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

/**
 * Title: 词频统计
 * Description: 
 * Date 2020/12/16
 */
object WordCountSQL {
  def main(args: Array[String]): Unit = {
    // 1.env
    val env = ExecutionEnvironment.getExecutionEnvironment
    val tEnv = BatchTableEnvironment.create(env)

    // 2.source
    val input = env.fromElements(WC("hello", 1), WC("hello", 1), WC("ciao", 1))

    // 3.transformation
    // register the DataSet as a view "WordCount"
    tEnv.createTemporaryView("WordCount", input, $"word", $"frequency")

    // run a SQL query on the Table and retrieve the result as a new Table
    val table = tEnv.sqlQuery("SELECT word, SUM(frequency) FROM WordCount GROUP BY word")

    // 4.sink
    table.toDataSet[WC].print()
  }

  case class WC(word: String, frequency: Long)

}
